package cn.com.idmy.mq;

import cn.com.idmy.mq.builder.MqMsgBuilder;
import cn.com.idmy.mq.builder.MqSendResultBuilder;
import cn.com.idmy.mq.coder.FastjsonMqMsgDecoder;
import cn.com.idmy.mq.coder.FastjsonMqMsgEncoder;
import cn.com.idmy.mq.coder.MqMsgEncoder;
import cn.com.idmy.mq.model.MqMsg;
import cn.com.idmy.mq.model.MqSendResult;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

@Configuration
@EnableConfigurationProperties({MqProps.class})
@RequiredArgsConstructor
class MqAutoConfiguration implements ApplicationContextAware {
    private Environment environment;

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        environment = ctx.getEnvironment();
    }

    private final MqProps props;

    @Bean
    @ConditionalOnMissingBean({FastjsonMqMsgDecoder.class})
    public FastjsonMqMsgDecoder fastjsonMqMsgDecoder() {
        return new FastjsonMqMsgDecoder();
    }

    @Bean
    @ConditionalOnMissingBean({FastjsonMqMsgEncoder.class})
    public FastjsonMqMsgEncoder fastjsonMqMsgEncoder() {
        return new FastjsonMqMsgEncoder();
    }

    @ConditionalOnExpression("${mq.enableConsumer:false}")
    @Bean
    public MqConsumerManager mqConsumerManager(FastjsonMqMsgDecoder decoder) {
        return new MqConsumerManager(props, decoder);
    }

    @ConditionalOnExpression("${mq.enableProducer:false}")
    @Bean
    public IMqProducer mqProducer(FastjsonMqMsgEncoder encoder) {
        Assert.notBlank(props.getClientType(), "clientType不能为空");
        RocketmqMsgBuilder msgBuilder = new RocketmqMsgBuilder();
        RocketmqSendResultBuilder sendResultBuilder = new RocketmqSendResultBuilder();
        RocketmqProducer producer = new RocketmqProducer(props, encoder, msgBuilder, sendResultBuilder);
        MqUtil.producer = producer;
        producer.start();
        return producer;
    }

    private class RocketmqMsgBuilder implements MqMsgBuilder<Message> {
        @Override
        public Message build(MqMsg msg, MqMsgEncoder encoder) {
            String topic = getTopic(msg.getTopic());
            Message rocketmqMsg = new Message(topic, msg.getTag(), msg.getKey(), msg.getMsgEncoder(encoder).encode(msg.getBody()));
            if (msg.getDelayLevel() != null) {
                rocketmqMsg.setDelayTimeLevel(msg.getDelayLevel().getLevel());
            }
            return rocketmqMsg;
        }
    }

    private String getTopic(String topic) {
        if (topic.startsWith("${") && topic.endsWith("}")) {
            return environment.resolveRequiredPlaceholders(topic);
        } else {
            return StrUtil.blankToDefault(topic, props.getTopic());
        }
    }

    private static class RocketmqSendResultBuilder implements MqSendResultBuilder<SendResult> {
        @Override
        public MqSendResult build(SendResult in) {
            MqSendResult out = new MqSendResult();
            if (in != null) {
                out.setMsgId(in.getMsgId());
                out.setTopic(in.getMessageQueue().getTopic());
            }
            return out;
        }
    }
}
